package tech.smartboot.redisun;

import org.smartboot.socket.transport.AioQuickClient;
import org.smartboot.socket.transport.AioSession;
import org.smartboot.socket.transport.MultiplexClient;
import tech.smartboot.redisun.cmd.DelCommand;
import tech.smartboot.redisun.cmd.HelloCommand;
import tech.smartboot.redisun.cmd.ZAddCommand;
import tech.smartboot.redisun.resp.Integers;
import tech.smartboot.redisun.resp.RESP;
import tech.smartboot.redisun.resp.SimpleErrors;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;

/**
 * @author 三刀
 * @version v1.0 10/21/25
 */
public final class Redisun extends MultiplexClient<RESP> {
    private final RedisunOptions options = new RedisunOptions(multiplexOptions);

    public static Redisun create(Consumer<RedisunOptions> opts) {
        Redisun redisun = new Redisun();
        opts.accept(redisun.options);
        return redisun;
    }

    private Redisun() {
        RedisMessageProcessor processor = new RedisMessageProcessor();
        multiplexOptions.init(processor, processor);
    }

    public int zadd(String key, double score, String member) {
        RESP r = syncExecute(new ZAddCommand(key, score, member));
        if (r instanceof Integers) {
            return ((Integers) r).getValue();
        }
        throw new RedisunException("invalid response:" + r);
    }

    public int del(String... keys) {
        return del(Arrays.asList(keys));
    }

    public int del(List<String> keys) {
        RESP r = syncExecute(new DelCommand(keys));
        if (r instanceof Integers) {
            return ((Integers) r).getValue();
        }
        throw new RedisunException("invalid response:" + r);
    }


    private RESP syncExecute(RedisSerialization command) {
        RESP resp;
        try {
            resp = execute(command).get();
        } catch (Throwable e) {
            throw new RedisunException(e);
        }
        if (resp instanceof SimpleErrors) {
            throw new RedisunException(((SimpleErrors) resp).getValue());
        }
        return resp;
    }

    private CompletableFuture<RESP> execute(RedisSerialization command) {
        CompletableFuture<RESP> future = new CompletableFuture<>();
        future.exceptionally(t -> {
            t.printStackTrace();
            return null;
        });
        try {
            AioQuickClient client = acquireClient();
            AioSession session = client.getSession();
            RedisSession redisSession = session.getAttachment();
            future.whenComplete((redisResponse, throwable) -> {
                redisSession.setFuture(null);
                redisSession.setDecodingResponse(null);
                reuseClient(client);
            });
            future.exceptionally(throwable -> {
                releaseClient(client);
                return null;
            });

            if (redisSession.getFuture() != null || redisSession.getDecodingResponse() != null) {
                throw new IllegalStateException("当前session正在执行其他命令");
            }
            redisSession.setFuture(future);
            synchronized (client) {
                command.writeTo(session.writeBuffer());
                session.writeBuffer().flush();

            }
        } catch (Throwable e) {
            future.completeExceptionally(e);
        }
        return future;
    }

    @Override
    protected void onNewClient(AioQuickClient client) {
        HelloCommand helloCommand = new HelloCommand();
        helloCommand.setUsername(options.getUsername());
        helloCommand.setPassword(options.getPassword());
        RESP redisResponse = syncExecute(helloCommand);
        HelloCommand.Response response = HelloCommand.toResponse(redisResponse);
        System.out.println(response);
    }

    @Override
    protected void onReuseClient(AioQuickClient client) {
        AioSession session = client.getSession();
        RedisSession attachment = session.getAttachment();
        //重置附件，为下一个响应作准备
        attachment.setDecodingResponse(null);
    }

}
